package concurrent

import (
	"context"
	"sync"
)

// Channel 定义任何类型通道
type Channel[T any] struct {
}

// NewChannel 实例化通道
func NewChannel[T any]() *Channel[T] {
	return &Channel[T]{}
}

// CreateChannel 基于输入值创建通道
// ctx 上下文
// values 输入值
// 返回值: 输出通道
func (c *Channel[T]) CreateChannel(ctx context.Context, values ...T) <-chan T {
	dataStream := make(chan T)
	go func() {
		defer close(dataStream)
		for _, value := range values {
			select {
			case <-ctx.Done():
				return
			case dataStream <- value:
			}
		}
	}()
	return dataStream
}

// Repeat 重复输出值
// ctx 上下文
// values 输入值
// 返回值: 输出通道
func (c *Channel[T]) Repeat(ctx context.Context, values ...T) <-chan T {
	dataStream := make(chan T)
	go func() {
		defer close(dataStream)
		for {
			for _, v := range values {
				select {
				case <-ctx.Done():
					return
				case dataStream <- v:
				}
			}
		}
	}()
	return dataStream
}

// RepeatFn 函数式重复输出值
// ctx 上下文
// fn 函数
// 返回值: 输出通道
func (c *Channel[T]) RepeatFn(ctx context.Context, fn func() T) <-chan T {
	dataStream := make(chan T)
	go func() {
		defer close(dataStream)
		for {
			select {
			case <-ctx.Done():
				return
			case dataStream <- fn():
			}
		}
	}()
	return dataStream
}

// Take 获取指定数量的值
// ctx 上下文
// valueStream 输入通道
// 返回值: 输出通道
func (c *Channel[T]) Take(ctx context.Context, valueStream <-chan T, limitNumber int) <-chan T {
	takeStream := make(chan T)
	go func() {
		defer close(takeStream)
		for i := 0; i < limitNumber; i++ {
			select {
			case <-ctx.Done():
				return
			case takeStream <- <-valueStream:
			}
		}
	}()
	return takeStream
}

// Merge 合并多个通道
// ctx 上下文
// channels 输入通道
// 返回值: 输出通道
func (c *Channel[T]) Merge(ctx context.Context, channels ...<-chan T) <-chan T {
	out := make(chan T)
	go func() {
		var wg sync.WaitGroup
		wg.Add(len(channels))
		for _, c := range channels {
			go func(ch <-chan T) {
				defer wg.Done()
				for v := range c {
					select {
					case <-ctx.Done():
						return
					case out <- v:
					}
				}
			}(c)
		}
		wg.Wait()
		close(out)
	}()
	return out
}

// Split 将一个通道拆分为两个通道
// ctx 上下文
// in 输入通道
// 返回值: 输出通道1, 输出通道2
func (c *Channel[T]) Split(ctx context.Context, in <-chan T) (<-chan T, <-chan T) {
	outChan1 := make(chan T)
	outChan2 := make(chan T)
	go func() {
		defer close(outChan1)
		defer close(outChan2)
		for val := range c.OrDone(ctx, in) {
			var out1, out2 = outChan1, outChan2
			for i := 0; i < 2; i++ {
				select {
				case <-ctx.Done():
					return
				case out1 <- val:
					out1 = nil
				case out2 <- val:
					out2 = nil
				}
			}
		}
	}()
	return outChan1, outChan2
}

// Or 监听多个通道
// channels 输入通道
// 返回值: 输出通道
// 监听多个通道，返回第一个可用的通道
func (c *Channel[T]) Or(channels ...<-chan T) <-chan T {
	length := len(channels)
	if length == 0 {
		return nil
	}
	if length == 1 {
		return channels[0]
	}
	moreDone := make(chan T)
	go func() {
		defer close(moreDone)
		if length == 2 {
			select {
			case <-channels[0]:
			case <-channels[1]:
			}
			return
		}
		select {
		case <-channels[0]:
		case <-channels[1]:
		case <-channels[2]:
		case <-c.Or(append(channels[3:], moreDone)...):
		}
	}()
	return moreDone
}

// OrDone 监听通道
// ctx 上下文
// channel 输入通道
// 返回值: 输出通道
func (c *Channel[T]) OrDone(ctx context.Context, channel <-chan T) <-chan T {
	valStream := make(chan T)
	go func() {
		defer close(valStream)
		for {
			select {
			case <-ctx.Done():
				return
			case v, ok := <-channel:
				if !ok {
					return
				}
				select {
				case valStream <- v:
				case <-ctx.Done():
				}
			}
		}
	}()
	return valStream
}

// Bridge 将多个通道合并成一个通道
// ctx 上下文
// chanStream 输入通道
// 返回值: 输出通道
func (c *Channel[T]) Bridge(ctx context.Context, chanStream <-chan <-chan T) <-chan T {
	valStream := make(chan T)

	go func() {
		defer close(valStream)

		for {
			var stream <-chan T
			select {
			case maybeStream, ok := <-chanStream:
				if !ok {
					return
				}
				stream = maybeStream
			case <-ctx.Done():
				return
			}

			for val := range c.OrDone(ctx, stream) {
				select {
				case valStream <- val:
				case <-ctx.Done():
				}
			}
		}
	}()

	return valStream
}
